/*
 * ***** BEGIN LICENSE BLOCK *****
 * Version: MPL 1.1
 *
 * The contents of this file are subject to the Mozilla Public License Version
 * 1.1 (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 * http://www.mozilla.org/MPL/
 *
 * Software distributed under the License is distributed on an "AS IS" basis,
 * WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
 * for the specific language governing rights and limitations under the
 * License.
 *
 * The Original Code is
 *
 * java-concurrency-presentation:
 * concurrency presentation tutorial.
 *
 * The Initial Developer of the Original Code is
 * Marco Antonio Villalobos, Jr.
 * Portions created by the Initial Developer are Copyright (C) 2010
 * the Initial Developer. All Rights Reserved.
 *
 * Contributor(s): None.
 *
 * ***** END LICENSE BLOCK *****
 */

package presentation.concurrency.indexer.combined;

import presentation.concurrency.indexer.Index;
import presentation.concurrency.indexer.Indexer;
import presentation.concurrency.indexer.Location;
import presentation.concurrency.indexer.Word;
import presentation.concurrency.util.ContextSwitcher;

import java.io.File;
import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * An example of how to incorrectly use concurrency.
 */
public class SafeCombinedApproachIndexer extends Indexer {

    private class IndexImpl implements Index {
        private final ContextSwitcher contextSwitcher = new ContextSwitcher();
        private final ConcurrentMap<String, Word> values = new ConcurrentHashMap<String, Word>();
        private final AtomicLong wordCount = new AtomicLong(0);
        private final ReadWriteLock rwl = new ReentrantReadWriteLock();
        private final Lock rl = rwl.readLock();
        private final Lock wl = rwl.writeLock();

        public void add(String word, String file, int line) {
            Word current = new WordImpl();
            current.add(file, line);

            wl.lock();
            try {
                Word previous = values.putIfAbsent(word, current);
                if (previous!=null) {
                    previous.add(file, line);
                }

                wordCount.incrementAndGet();
                contextSwitcher.forceSometimes();
            } finally {
                wl.unlock();
            }
        }

        public Set<Map.Entry<String, Word>> getWords() {
            rl.lock();
            try {
                return new TreeMap<String, Word>(values).entrySet();
            } finally {
                rl.unlock();
            }
        }

        public long getWordCount() {
            rl.lock();
            try {
                return wordCount.get();
            } finally {
                rl.unlock();
            }
        }
    }


    private class WordImpl implements Word {
        private final AtomicLong count = new AtomicLong(0);
        private final Set<Location> locations = new LinkedHashSet<Location>();
        private final ReadWriteLock rwl = new ReentrantReadWriteLock();
        private final Lock rl = rwl.readLock();
        private final Lock wl = rwl.writeLock();

        public void add(String file, int line) {
            wl.lock();
            try {
                Location key = new Location(file, line);
                locations.add(key);
                count.incrementAndGet();
            } finally {
                wl.unlock();
            }
        }

        public long getCount() {
            rl.lock();
            try {
                return count.get();
            } finally {
                rl.unlock();
            }
        }

        public Set<Location> getLocations() {
            rl.lock();
            try {
                return new LinkedHashSet<Location>(locations);
            } finally {
                rl.unlock();
            }
        }
    }

    public Index execute(final File[] files) {
        final int no_of_threads = 25;
        final ArrayBlockingQueue<File> fileProducer = new ArrayBlockingQueue<File>(files.length);
        final CountDownLatch latch = new CountDownLatch(files.length);
        Collections.addAll(fileProducer, files);

        final ExecutorService pool = Executors.newFixedThreadPool(no_of_threads);

        final Index index = new IndexImpl();

        Thread fileConsumer = new Thread() {
            public void run() {

                try {
                    while (!Thread.currentThread().isInterrupted()) {

                        final File file = fileProducer.take();
                        Runnable indexer = new Runnable() {
                            public void run() {
                                index(index, file);
                                latch.countDown();
                            }
                        };
                        pool.submit(indexer);
                    }
                }
                catch (InterruptedException e) {
                    //allow for exit
                }
            }
        };
        fileConsumer.start();

        try {
            latch.await();
        } catch (InterruptedException e) {
            //signal that all files were processed
        }
        pool.shutdown();
        try {
            pool.awaitTermination(Integer.MAX_VALUE, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            //won't be interrupted
        }
        fileConsumer.interrupt();

        return index;
    }
}